//package com.kj.report.utils;
//
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQBrokerException;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.client.producer.DefaultMQProducer;DefaultMQProducer
//import org.apache.rocketmq.client.producer.SendResult;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.Message;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.remoting.exception.RemotingException;
//
//import java.io.UnsupportedEncodingException;
//import java.util.List;
//
//public class RocketMQUtils {
//    //消息发送
//    public static SendResult  mqProducer(String topic,String tag,String message) {
//        DefaultMQProducer producer = new DefaultMQProducer("KJ_P_REPORT");
//        //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
//        producer.setNamesrvAddr("192.168.10.88:9876");
//        SendResult sendResult = null;
//        try {
//            //启动producer实例
//            producer.start();
//
//            Message messages = new Message(topic, tag, message.getBytes("utf-8"));
//            sendResult = producer.send(messages);
//
//        } catch (MQClientException e) {
//            e.printStackTrace();
//        } catch (UnsupportedEncodingException e) {
//            e.printStackTrace();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        } catch (RemotingException e) {
//            e.printStackTrace();
//        } catch (MQBrokerException e) {
//            e.printStackTrace();
//        }
//
//        producer.shutdown();
//        return sendResult;
//    }
//
//
//    public static String consumer(String topic,String tag){
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("KJ_C_REPORT");
//        //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
//        consumer.setNamesrvAddr("192.168.10.88:9876");
//        //这里设置的是一个consumer的消费策略
//        //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
//        //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
//        //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//
//
//        try {
//            consumer.subscribe(topic, tag);
//
//            //设置consumer所订阅的Topic和Tag，*代表全部的Tag
//            consumer.registerMessageListener(new MessageListenerConcurrently() {
//                @Override
//                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msg, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//
//                    msg.stream().forEach(param ->{
//                        String messageBody = new String(param.getBody());
//
//                        System.out.println("消费响应：msgId : " + param.getMsgId() + ",  msgBody : " + messageBody);//输出消息内容
//                    });
//                    //返回消费状态
//                    // CONSUME_SUCCESS 消费成功
//                    // RECONSUME_LATER 消费失败，需要稍后重新消费
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//
//                }
//            });
//            consumer.start();
//        } catch (MQClientException e) {
//            e.printStackTrace();
//        }
//        //调用start()方法启动consumer
//
//
//        System.out.println("Consumer Started.");
//        return "";
//    }
//}
